package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MQException;
import com.example.mq.mqserver.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 实现消费者消费消息的核心逻辑
 */
@Slf4j
public class ConsumerManager {
    // 持有上层的 VirtualHost 对象的引用，用来操作数据
    private final VirtualHost parent;
    // 创建一个线程池，负责去执行具体的回调任务
    private final ExecutorService workerPool = Executors.newCachedThreadPool();
    // 存放令牌（queueName）的阻塞队列
    private final BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();

    public ConsumerManager (VirtualHost p) {
        parent = p;

        // 扫描线程
        // 1. 从阻塞队列中获取令牌
        // 2. 根据令牌，找到指定的队列
        // 3. 从这个队列中消费一个消息
        Thread scannerThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 从阻塞队列中获取令牌
                    String queueName = tokenQueue.take();
                    // 2. 根据令牌，找到指定的队列
                    MQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MQException("[VirtualHost] 获取令牌后，该队列不存在！queueName="
                                + queueName);
                    }
                    // 3. 从这个队列中消费一个消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MQException e) {
                    e.printStackTrace();
                }
            }
        });
        // 将这个线程设置为后台线程，后台线程不会影响到进程的结束
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    // 发送完消息（队列收到消息），就需要调用这个方法，将令牌（队列名）放入阻塞队列
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }



    /**
     * 消费者订阅队列
     * 将消费者添加到指定队列到消费者队列
     * @param consumerTag   消费者身份标识
     * @param queueName     队列名
     * @param autoAck       是否自动应答
     * @param consumer      回调函数
     */
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MQException {
        // 找到指定的队列
        MQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MQException("[VirtualHost] 队列不存在！queueName=" + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有了一些消息，需要立即就消费掉
            // 这种情况是消息积累了，可能是之前没有消费者（这是第一个消费者），或者之前消息消费速度跟不上生产
            int n = parent.getMemoryDataCenter().getMessageCount(queue.getName());
            for (int i = 0; i < n; i++) {
                try {
                    // 将队列名放入阻塞队列中，让扫描线程去扫描并消费消息
                    notifyConsume(queueName);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 消费消息
     * 从指定队列中消费一个消息
     * @param queue 队列
     */
    public void consumeMessage(MQueue queue) {
        // 1. 按照轮询的方式，在队列中挑选一个消费者
        ConsumerEnv luckyDog = queue.chooseConsumerEnv();
        if (luckyDog == null) {
            // 当前队列没有消费者（订阅者）
            return;
        }
        // 2. 从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            // 当前队列中还没有消息，不需要消费。
            return;
        }
        // 3. 把消息放到消费者的回调方法中，交给线程池执行。
        workerPool.submit(() -> {
            try {
                // 1. 将消息放到待确认消息表
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 2. 执行回调函数
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());

                // 3. 如果是自动应答，就可以删除消息了，如果是手动应答就不需要处理（回调函数中调用 basicAck 进行处理）
                if (luckyDog.isAutoAck()) {
                    // 1) 删除硬盘上的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message.getMessageId(), parent.getMemoryDataCenter());
                    }
                    // 2) 删除内存的消息中心上的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    // 3) 删除待确认消息表的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    log.info("消息被成功消费！queueName=" + queue.getName());
                }
            } catch (IOException | ClassNotFoundException | MQException e) {
                e.printStackTrace();
            }
        });
    }
}
